package com.renrendoc.network.flow.source;

import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.util.Timer;
import java.util.TimerTask;

public abstract class ConfigSource<OUT> extends RichSourceFunction<OUT> {
    boolean isRunning = true;
    private boolean trick = true;
    private Long scheduleTime = 30000L;

    public ConfigSource() {

    }

    public ConfigSource(Long t) {
        scheduleTime = t;
    }

    public void setScheduleTime(Long t) {
        scheduleTime = t;
    }

    public abstract OUT reFresh();

    @Override
    public void run(SourceContext<OUT> ctx) throws Exception {
        ctx.collect(reFresh());
        while (isRunning) {
//            TimeUnit.SECONDS.sleep(1);
//            ctx.collect(reFresh());
            if (trick) {
                trick = false;
                new Timer().schedule(new TimerTask() {
                    @Override
                    public void run() {
                        trick = true;
                        ctx.collect(reFresh());
                    }
                }, scheduleTime, scheduleTime);
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
